package com.atguigu.realtime.utils;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * Created by Smexy on 2022/7/18
 *
 *      单例。
 *
 *   -----------------------
 *      单例怎么写?
 *            static  producer
 *
 *            static {
 *
 *         //为静态属性赋值
 *         producer = createProducer();
 *
 *          }
 *      不是单例的怎么写?
 *
 *         以分区为单位去new KafkaProducerUtil2
 *         KafkaProducerUtil2 client =     new KafkaProducerUtil2()
 *         client.sendData()
 *
 *   --------------------------
 *   什么时候需要设置为单例?
 *          如果希望一个JVM中只能new 一个(一次)这个对象，设置为单例。
 *          典型的是哪些消耗资源，消耗时间的操作。
 *          举例：  HBase ：  Connection
 *                  一个App有一个 Connection，就可以设计为单例
 *
 *    ---------------------------
 *          当前KafkaProducerUtil 是单例和不是单例有什么区别?
 *
 *                  以分区为单位使用 生产者发送数据。
 *
 *                  功能上没区别，非单例的，需要创建更多的Producer，需要更多的资源(缓冲区)
 *
 *
 *
 */
public class KafkaProducerUtil2 {

    private  Producer<String,String> producer;

     {

        //为静态属性赋值
        producer = createProducer();

    }

    private  Producer<String,String> createProducer(){

        Properties properties = new Properties();

        /*
                必须的参数:
                        集群地址
                        Key,value的序列化器

                 参考 ProducerConfig
         */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,PropertiesUtil.getProperty("kafka.broker.list"));
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,PropertiesUtil.getProperty("key.serializer"));
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,PropertiesUtil.getProperty("value.serializer"));
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");

        //提供生产幂等性

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        return producer;

    }

    // 生产指定的消息到指定的topic
    public  void sendData(String message,String topic){

        producer.send(new ProducerRecord<String,String>(topic,message));


    }

    // 生产的数据先进入buffer，满足一定条件(缓冲区大小，时间间隔)才会发送到broker
    public  void flush(){
        producer.flush();
    }

}
